package com.jeffse.light.stream.core.function.base;

import com.jeffse.light.stream.core.data.context.Context;
import com.jeffse.light.stream.core.data.descriptor.operatorDescriptor.OperatorDescriptor;
import com.jeffse.light.stream.core.pojo.OutTag;

public abstract class ProcessFunction<T, O> implements AbstractProcessFunction<T,O> {
    private Context ctx;
    private OutTag outTag;
    private OperatorDescriptor nextOperatorDescriptor = null;

    @Override
    public void setNextOperatorDescriptor(OperatorDescriptor nextOperatorDescriptor) {
        this.nextOperatorDescriptor = nextOperatorDescriptor;
    }

    @Override
    public OperatorDescriptor getNextOperatorDescriptor() {
        return nextOperatorDescriptor;
    }

    @Override
    public OutTag getOutTag() {
        return outTag;
    }

    @Override
    public void setOutTag(OutTag outTag) {
        this.outTag = outTag;
    }

    @Override
    public void setCtx(Context ctx) {
        this.ctx = ctx;
    }

    @Override
    public Context getCtx() {
        return ctx;
    }

    @Override
    public void collect(O element, OutTag outTag) throws Exception {
        ctx.collect(element, outTag);
    }

    @Override
    public void collect(O element) throws Exception {
        ctx.collect(element, nextOperatorDescriptor);
    }
}
